<?php
namespace app\command;

use app\service\MqttClient;
use app\service\MqttMessageHandler;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Config;
use think\facade\Log;


class MqttClientCommand extends Command{
    protected $startTime;
    protected $messageCount = 0;
    protected $lastStatsTime;
    protected $topics = ['test/tp5' => 0];
    protected $reconnectCount = 0;
    protected $maxReconnects = 10;
    protected $reconnectDelay = 5; // 重连延迟（秒）
    
    /**
     * 消息处理器
     * @var MqttMessageHandler
     */
    protected $messageHandler;
    
    /**
     * 是否开启数据库存储
     * @var bool
     */
    protected $enableDbStorage = true;
    
    protected function configure()
    {
        $this->setName('mqtt:client')
             ->setDescription('订阅MQTT消息')
             ->addOption('topic', 't', \think\console\input\Option::VALUE_OPTIONAL, '订阅的主题(多个用逗号分隔)', 'tc/device/out/#,rziot/zftx_w06/#')
             ->addOption('quality', 'l', \think\console\input\Option::VALUE_OPTIONAL, '服务质量(0-2)', 0)
             ->addOption('no-db', null, \think\console\input\Option::VALUE_NONE, '禁用数据库存储');
    }

    protected function execute(Input $input, Output $output)
    {
        
    
        // 初始化性能监控
        $this->startTime = time();
        $this->lastStatsTime = $this->startTime;
        
        
        // 获取命令行参数
        $topicStr = $input->getOption('topic');
        $qos = (int)$input->getOption('quality');

        // 解析多个主题（用逗号分隔）
        $topicList = array_map('trim', explode(',', $topicStr));
        $this->topics = [];
        foreach ($topicList as $t) {
            if (!empty($t)) {
                $this->topics[$t] = $qos;
            }
        }

        $output->writeln('DEBUG: 获取命令行参数完成: topics=' . json_encode($this->topics) . ', qos=' . $qos);
        
        // 是否禁用数据库存储
        $this->enableDbStorage = !$input->getOption('no-db');
        $output->writeln('DEBUG: enableDbStorage=' . ($this->enableDbStorage ? 'true' : 'false'));
        
        // 初始化消息处理器
        
        if ($this->enableDbStorage) {
            
            try {
                $this->messageHandler = new MqttMessageHandler();
                //$output->writeln('DEBUG: 创建MqttMessageHandler实例成功');
                //$output->writeln('数据库存储已启用');
            } catch (\Throwable $e) {
                $output->writeln('ERROR: 创建MqttMessageHandler失败: ' . $e->getMessage());
                Log::error('创建MqttMessageHandler失败: ' . $e->getMessage() . "\n" . $e->getTraceAsString());
            }
        } else {
            $output->writeln('数据库存储已禁用');
        }
        
        // 获取配置
        $config = Config::get('mqtt.');
        
        // 确保配置中的数值类型正确
        $config['port'] = (int)$config['port'];
        $config['keepalive'] = (int)$config['keepalive'];
        
        // 动态生成client_id
        if (empty($config['client_id']) || strpos($config['client_id'], 'uniqid()') !== false) {
            $config['client_id'] = 'tp_client_' . uniqid();
        }
        
        $output->writeln('正在连接到MQTT服务器...');
        $output->writeln('配置: ' . json_encode($config, JSON_UNESCAPED_UNICODE));
        
        try {
            $mqtt = new MqttClient($config);

            // 添加重试逻辑
            $maxRetries = 3;
            $connected = false;

            for ($i = 0; $i < $maxRetries; $i++) {
                $output->writeln("尝试连接 (" . ($i + 1) . "/{$maxRetries})...");

                if ($mqtt->connect()) {
                    $connected = true;
                    break;
                } else {
                    $error = $mqtt->getLastError();
                    $output->writeln("连接失败: {$error}");

                    if ($i < $maxRetries - 1) {
                        $output->writeln("等待 2 秒后重试...");
                        sleep(2);

                        // 重新生成客户端ID
                        $config['client_id'] = 'tp_client_' . uniqid() . '_' . mt_rand(1000, 9999);
                        $mqtt = new MqttClient($config);
                    }
                }
            }

            if (!$connected) {
                $output->writeln('所有连接尝试都失败了，请检查EMQ X服务器状态');
                return;
            }
            
            $topicStr = implode(', ', array_keys($this->topics));
            $output->writeln('连接成功，订阅主题: ' . $topicStr . ' (QoS: ' . $qos . ')');
            $result = $mqtt->subscribe($this->topics);
            
            // 检查订阅结果
            if ($result === true || is_array($result)) {
                $output->writeln('订阅成功: ' . json_encode($result, JSON_UNESCAPED_UNICODE));
            } else {
                $output->writeln('订阅失败: ' . $mqtt->getLastError());
                return;
            }
            
            $output->writeln('等待消息... (按 Ctrl+C 退出)');
            
            // 注册信号处理
            if (function_exists('pcntl_signal')) {
                pcntl_signal(SIGINT, function() use ($mqtt, $output) {
                    $output->writeln("\n正在断开连接...");
                    $mqtt->disconnect();
                    
                    // 确保所有消息都被保存
                    if ($this->enableDbStorage && $this->messageHandler) {
                        $this->messageHandler->flush();
                    }
                    
                    $this->showStats($output);
                    exit(0);
                });
            }
            
            // 设置无限等待时间
            while (true) {
                try {
                    // 处理信号
                    if (function_exists('pcntl_signal_dispatch')) {
                        pcntl_signal_dispatch();
                    }
                    
                    // 检查连接状态
                    if (!$mqtt->isConnected()) {
                        $output->writeln('连接已断开，尝试重连...');
                        
                        if ($this->reconnectCount >= $this->maxReconnects) {
                            $output->writeln('达到最大重连次数，退出程序');
                            break;
                        }
                        
                        $this->reconnectCount++;
                        $output->writeln("第 {$this->reconnectCount} 次重连 (最大 {$this->maxReconnects} 次)");
                        
                        if (!$mqtt->reconnect()) {
                            $output->writeln('重连失败: ' . $mqtt->getLastError());
                            sleep($this->reconnectDelay);
                            continue;
                        }
                        
                        $output->writeln('重连成功，重新订阅主题');
                        $result = $mqtt->subscribe($this->topics);
                        
                        if (!($result === true || is_array($result))) {
                            $output->writeln('重新订阅失败: ' . $mqtt->getLastError());
                            sleep($this->reconnectDelay);
                            continue;
                        }
                        
                        $output->writeln('重新订阅成功，继续接收消息');
                        $this->reconnectCount = 0; // 重置重连计数
                    }
                    
                    $message = $mqtt->receive();
                    
                    // 处理接收结果
                    if ($message === false) {
                        // 接收超时或错误，但不一定是连接断开
                        // 短暂等待后继续
                        usleep(100000); // 100ms
                        continue;
                    }
                    
                    if (is_array($message)) {
                        $this->messageCount++;

                        // 记录消息摘要（避免日志过长）
                        $messageSummary = [
                            'type' => $message['type'] ?? 0,
                            'topic' => $message['topic'] ?? '未知',
                            'qos' => $message['qos'] ?? 0,
                            'has_message' => isset($message['message']),
                            'message_length' => isset($message['message']) && is_string($message['message']) ? strlen($message['message']) : 0
                        ];
                        Log::info("接收到MQTT消息: " . json_encode($messageSummary, JSON_UNESCAPED_UNICODE));

                        // 处理消息
                        $this->processMessage($message, $output);
                        
                        // 每接收100条消息或每60秒显示一次统计信息
                        if ($this->messageCount % 100 === 0 || (time() - $this->lastStatsTime) >= 60) {
                            $this->showStats($output);
                        }
                        
                        // 重置重连计数，因为成功接收了消息
                        $this->reconnectCount = 0;
                    } else {
                        //$output->writeln("收到未知格式消息: " . var_export($message, true));
                        //Log::warning("收到未知格式MQTT消息: " . var_export($message, true));
                        
                        // 尝试将未知格式消息转换为可处理的格式
                        if ($message === true) {
                            //$output->writeln("收到心跳确认消息，忽略处理");
                        } else if (is_string($message)) {
                            // 尝试解析为JSON
                            $jsonData = json_decode($message, true);
                            if (json_last_error() === JSON_ERROR_NONE && is_array($jsonData)) {
                                $output->writeln("成功将字符串转换为JSON消息");
                                // 构造标准消息格式
                                $standardMessage = [
                                    'topic' => 'unknown/topic',
                                    'message' => $message,
                                    'qos' => 0
                                ];
                               
                                $this->processMessage($standardMessage, $output);
                            } else {
                                $output->writeln("无法处理的字符串消息: " . $message);
                            }
                        }
                    }
                    
                    // 短暂休眠，减少CPU使用率
                    usleep(100000); // 10ms，可以根据需要调整
                } catch (\Throwable $e) {
                    $output->writeln("错误: " . $e->getMessage());
                    Log::error("MQTT客户端错误: " . $e->getMessage() . "\n" . $e->getTraceAsString());
                    
                    // 如果是致命错误，增加重连延迟
                    sleep($this->reconnectDelay);
                }
            }
            
            $mqtt->disconnect();
            
            // 确保所有消息都被保存
            if ($this->enableDbStorage && $this->messageHandler) {
                $this->messageHandler->flush();
            }
            
            $this->showStats($output);
            
        } catch (\Throwable $e) {
            $output->writeln('错误: ' . $e->getMessage());
            Log::error("MQTT客户端异常: " . $e->getMessage() . "\n" . $e->getTraceAsString());
            return;
        }
    }
    
    /**
     * 处理接收到的消息
     * @param array $message 消息数据
     * @param Output $output 输出对象
     */
    protected function processMessage(array $message, Output $output): void
    {
        $topic = $message['topic'] ?? '未知主题';
        $messageType = $message['type'] ?? 0;

        // 检查是否有消息内容
        if (!isset($message['message'])) {
            $output->writeln(date('Y-m-d H:i:s') . " 收到类型 [{$messageType}] 主题 [{$topic}] 的控制消息（无数据内容）");
            Log::info("收到MQTT控制消息: type={$messageType}, topic={$topic}");
            return; // 没有消息内容，直接返回
        }

        // 检查是否是设备数据主题（支持多种协议）
        // Vendor1: tc/device/out/*
        // Vendor2: rziot/zftx_w06/*/result
        $isDeviceTopic = (strpos($message['topic'], 'tc/device/out/') === 0) ||
                         (strpos($message['topic'], 'rziot/zftx_w06/') === 0);

        if (!isset($message['topic']) || !$isDeviceTopic) {
            $output->writeln(date('Y-m-d H:i:s') . " 收到非设备数据主题的消息，跳过处理: {$topic}");
            return;
        }

        $content = $message['message'];

        // 显示消息内容（截断显示，避免日志过长）
        $displayContent = is_string($content) ? (strlen($content) > 200 ? substr($content, 0, 200) . '...' : $content) : json_encode($content);
        $output->writeln(date('Y-m-d H:i:s') . " 收到主题 [{$topic}] 的数据消息: {$displayContent}");

        // 解析JSON消息内容
        if (is_string($content)) {
            // 首先尝试直接解析（可能不需要处理转义）
            $decodedContent = json_decode($content, true);

            // 如果直接解析失败，尝试处理转义字符
            if (json_last_error() !== JSON_ERROR_NONE) {
                $output->writeln("直接JSON解析失败，尝试处理转义字符...");
                $unescapedContent = stripcslashes($content);
                $decodedContent = json_decode($unescapedContent, true);
            }

            // 检查最终解析结果
            if (json_last_error() === JSON_ERROR_NONE && is_array($decodedContent)) {
                $content = $decodedContent;
                $protocolType = isset($content['hotspot_info_5g']) ? '5G' : '4G';
                $output->writeln("JSON解析成功，协议类型: {$protocolType}");

                // 输出格式化的JSON（用于调试）
                if ($output->isVerbose()) {
                    $output->writeln("格式化的消息内容:");
                    $output->writeln(json_encode($content, JSON_PRETTY_PRINT | JSON_UNESCAPED_UNICODE));
                }
            } else {
                $errorMsg = json_last_error_msg();
                $output->writeln("JSON解析失败: {$errorMsg}");

                // 检查是否是截断问题
                if ($errorMsg === 'Syntax error') {
                    $output->writeln("可能的原因：消息被截断或格式错误");
                    Log::warning("JSON消息可能被截断: topic={$topic}, 长度=" . strlen($content) . ", 结尾=" . substr($content, -50));
                } else {
                    Log::error("JSON解析失败: {$errorMsg}, 原始内容长度: " . strlen($content));
                }
                return; // 解析失败则不处理
            }
        } else {
            $output->writeln("消息内容不是字符串格式，跳过JSON解析");
            return;
        }
        // 如果启用了数据库存储，则处理消息
        if ($this->enableDbStorage && $this->messageHandler) {
            $output->writeln("正在处理消息并保存到数据库...");
            Log::info("MqttClientCommand: 准备处理消息并保存到数据库");
            
            try {
                $result = $this->messageHandler->handle($topic,$content);
                if (!$result) {
                    //$output->writeln("消息处理失败");
                    Log::warning("MQTT消息处理失败: " . json_encode($message, JSON_UNESCAPED_UNICODE));
                } else {
                    $output->writeln("消息处理成功，已添加到缓冲区");
                    // 强制立即保存，不等待缓冲
                    $this->messageHandler->flush();
                    $output->writeln("已强制保存消息到数据库");
                }
            } catch (\Throwable $e) {
                //$output->writeln("消息处理异常: " . $e->getMessage());
                Log::error("MQTT消息处理异常: " . $e->getMessage() . "\n" . $e->getTraceAsString());
            }
        } else {
            $output->writeln("数据库存储已禁用，跳过保存");
        }
    }
    
    /**
     * 显示统计信息
     */
    protected function showStats(Output $output)
    {
        $now = time();
        $runTime = $now - $this->startTime;
        $hours = floor($runTime / 3600);
        $minutes = floor(($runTime % 3600) / 60);
        $seconds = $runTime % 60;
        
        $timeStr = sprintf('%02d:%02d:%02d', $hours, $minutes, $seconds);
        $msgPerSecond = $runTime > 0 ? round($this->messageCount / $runTime, 2) : 0;
        
        $output->writeln("-----------------------------------");
        $output->writeln("运行时间: {$timeStr}");
        $output->writeln("接收消息: {$this->messageCount} 条");
        $output->writeln("平均速率: {$msgPerSecond} 条/秒");
        $output->writeln("重连次数: {$this->reconnectCount}");
        $output->writeln("数据库存储: " . ($this->enableDbStorage ? '已启用' : '已禁用'));
        $output->writeln("-----------------------------------");
        
        $this->lastStatsTime = $now;
    }
}